package com.dh.cb;

import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public class ConsumereThread extends Thread {

    List<LoadHandler> handlers;

    AtomicInteger batch = new AtomicInteger(0);

    public void setHandlers(List<LoadHandler> handlers) {
        this.handlers = handlers;
    }

    @Override
    public void run() {
        System.out.println("consumer thread start");
        while (true) {
            try {
                List<String> msgList = null;
                try {
                    msgList = MsgContainer.queue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("consumer poll success");
                //Hash分组
                Map<Integer, List<String>> map =
                        msgList.stream().collect(Collectors.groupingBy(t -> Math.abs(t.hashCode()) % Constant.LOAD_THREAD));
                map.forEach((index, rows) -> {
                    LoadHandler handler = handlers.get(index);
                 //   System.out.println("index--" + index + "----" + handler.getTasks().size());
                    handler.offer(rows, index);
                  //  System.out.println("index--offer--end");
                });
                batch.incrementAndGet();
            }catch (Exception e){
                e.printStackTrace();
            }

        }
    }

    public void saveOffset() {
        System.out.println("\n\n-----save " + batch.get() + " offset is exe------\n\n");
    }

}
